package com.littlerpc.protocol.net.netty.client;


import com.littlerpc.protocol.codec.Serializer;
import com.littlerpc.protocol.net.common.*;
import com.littlerpc.protocol.net.netty.codec.NettyDecoder;
import com.littlerpc.protocol.net.netty.codec.NettyEncoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.Setter;

import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 *
 * @author ChengPeng
 * @date 2017/6/13
 *  这个类是在配置文件里面配置
 */

@Setter
public class NettyClient extends RpcClient {

    public NettyClient(String remoteHost,int remoteport, Serializer serializer) {
        super( remoteHost,remoteport,serializer);
        init();
    }

    /**
     * 保存客户端的请求，便于后续处理
     */
    private Map<String, Object> proxyPool = new ConcurrentHashMap<>();

    private Channel channel = null;


    @Override
    public Object send(RpcRequest request) {
        System.out.println("chanel***********" + channel);
        //构造RpcFuture
        RpcFuture<RpcResponse> future = RpcFuturePool.createFuture(request.getRequestId());
        //写入到服务端
        channel.writeAndFlush(request);

        try {
            RpcResponse response = future.get();
            return response.getResult();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            RpcFuturePool.removeFuture(request.getRequestId());
        }

        return null;
    }

    /**
     * 用于保存链接服务端的链接
     */
    private static Map<String, NettyClient> clientPool = new ConcurrentHashMap<>();


    /**
     * 初始化客户端连接
     */
    private void init() {
        EventLoopGroup group = new NioEventLoopGroup(1);
        Bootstrap client = new Bootstrap();
        client.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_REUSEADDR, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new NettyDecoder(serializer, RpcResponse.class));
                        pipeline.addLast(new NettyEncoder(serializer, RpcRequest.class));
                        pipeline.addLast(new NettyClientHander());
                    }
                });

        try {

            ChannelFuture f = client.connect(new InetSocketAddress(remoteHost, remotePort)).sync();
            channel = f.channel();
//            f.channel().closeFuture().sync();

        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}
